﻿#include "DataReceiver.h"

/// @brief 接收数据任务线程
void DataReceiver::taskRecv()
{
    int countA = 0, countB = 0, countC = 0; // test

    zmq_pollitem_t items[] = { { pull_sock_, 0, ZMQ_POLLIN, 0 } };
    while (sampling_) {
        zmq_poll(items, 1, 0); // 轮询，等待0毫秒（一直轮询），-1表示一直阻塞
        if ((items[0].revents & ZMQ_POLLIN) != 0) {
            size_t size = 0;
            char *contents = s_recv((void *)pull_sock_, size);
            if (contents == nullptr) {
                continue;
            }
            int head = static_cast<int>(contents[0]);
            // LOG_INFO("size = {}, head = {}", size, head);

            TotalMessage total_message;
            total_message.ParseFromArray(contents + 1, size);

            switch (head) {
            case TotalMessage::KFreqA: {
                LOG_INFO("freqA enqueue:{}.", countA++);
                message_queue_freqA_->enqueue(std::move(total_message));
                break;
            }
            case TotalMessage::KFreqB: {
                LOG_INFO("freqB enqueue:{}.", countB++);
                message_queue_freqB_->enqueue(std::move(total_message));
                break;
            }
            case TotalMessage::KFreqC: {
                LOG_INFO("freqC enqueue:{}.", countC++);
                message_queue_freqC_->enqueue(std::move(total_message));
                break;
            }
            }
            if (record_) {
                // 从客户端收到数据写文件
                int msg_size = total_message.sensor_data_size(); // 对应采集了多少次
                for (int i = 0; i < msg_size; i++) {
                    TotalMessage::SensorData sensor_data = total_message.sensor_data(i); // 获取子消息
                    int sensors_count = sensor_data.sensor_id_size();

                    for (int j = 0; j < sensors_count; j++) {
                        logfile_recv_->info("freq_type = {}, timestamp = {}, sensor_id = {}, value = {}.",
                                            total_message.freq_type(), sensor_data.timestamp(j),
                                            sensor_data.sensor_id(j), sensor_data.value(j));
                    }
                }
            }
            free(contents);
        }

        std::this_thread::sleep_for(std::chrono::microseconds(1));
    }
}

/// @brief 发送开始采集的指令
bool DataReceiver::startSampling(const int &freqa, const int &freqb, const int &freqc, const int &sensors_num,
                                 const bool &record)
{
    record_ = record;
    // 发送命令（开始采集）到数据采集端
    Command cmd;
    cmd.set_command_type(Command::KStartSampling);
    cmd.set_freqa(freqa);
    cmd.set_freqb(freqb);
    cmd.set_freqc(freqc);
    cmd.set_sensors_num(sensors_num);
    cmd.set_record(record);

    size_t len = cmd.ByteSizeLong() + 1;
    char *buff = new char[len];
    memset(buff, 0, len);
    buff[0] = Command::KStartSampling;
    cmd.SerializeToArray(buff + 1, len - 1);

    int ret = s_send((void *)push_sock_, buff, len, ZMQ_DONTWAIT);
    delete[] buff;
    if (-1 == ret) {
        return false;
    } else {
        return true;
    }
}

/// @brief 发送停止采集的指令
bool DataReceiver::stopSampling()
{
    // 发送命令（停止采集）到数据采集端
    size_t size = 1;
    char *buff = new char[size];
    memset(buff, 0, 1);
    buff[0] = Command::KStopSampling;

    int ret = s_send((void *)push_sock_, buff, size, ZMQ_DONTWAIT);
    delete[] buff;
    if (-1 == ret) {
        return false;
    } else {
        return true;
    }
}

/// @brief 停止线程
void DataReceiver::stopTask()
{
    // 线程标识置为false并等待线程结束
    sampling_ = false;
    if (thread_recv_.joinable()) {
        thread_recv_.join();
    }
    // 线程结束后再释放其他资源
    if (message_queue_freqA_) {
        delete message_queue_freqA_;
    }
    if (message_queue_freqB_) {
        delete message_queue_freqB_;
    }
    if (message_queue_freqC_) {
        delete message_queue_freqC_;
    }

    pull_sock_.close();
    push_sock_.close();
    context_.close();
}

bool DataReceiver::handlData(TotalMessage &&total_message, const TotalMessage::FreqType &freq_type)
{
    LOG_INFO("DataReceiver, Queue_freqA Over:{},Queue_freqB Over:{},Queue_freqC Over:{}", message_queue_freqA_->size(),
             message_queue_freqB_->size(), message_queue_freqC_->size());

    switch ((int)freq_type) {
    case TotalMessage::KFreqA: {
        return message_queue_freqA_->try_dequeue(std::move(total_message));
    }
    case TotalMessage::KFreqB: {
        return message_queue_freqB_->try_dequeue(std::move(total_message));
    }
    case TotalMessage::KFreqC: {
        return message_queue_freqC_->try_dequeue(std::move(total_message));
    }
    }
}